/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.nmpp.handler;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
import cn.ac.iie.di.ban.nmpp.metastore.TableSe2DBRule;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.util.HashMap;
import java.util.Map;
import static cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv.ROCKETMQ_DOC_DESC;
import java.util.Calendar;

/**
 *
 * @author alexmu
 */
public class TableSe2DWHandler {

    TableSe2DBRule tableSe2DBRule = null;
    List<SE2DWWorker> se2DBWorkerSet = new ArrayList<SE2DWWorker>();
    ExecutorService threadPool = Executors.newFixedThreadPool(100);
    private BlockingQueue<List> dataCache = new ArrayBlockingQueue<List>((Integer) RuntimeEnv.getParam(RuntimeEnv.DATA_CACHE_SIZE));
    private AtomicLong loadVolume = new AtomicLong();
    public static Map<String, REReceiveSession> consumerSet = new HashMap<String, REReceiveSession>();
    static Object consumerLock = new Object();
    static Map<String, TableSe2DWHandler> tableSe2DWHandlerSet = new HashMap<String, TableSe2DWHandler>();
    /*
     * 
     */
    private static Logger logger = Logger.getLogger(TableSe2DWHandler.class.getName());

    private TableSe2DWHandler(TableSe2DBRule pTableSe2DBRule) {
        tableSe2DBRule = pTableSe2DBRule;
    }

    public static TableSe2DWHandler getTableSe2DBHandler(TableSe2DBRule pTableSe2DBRule) {
        TableSe2DWHandler tableSe2DBHandler = new TableSe2DWHandler(pTableSe2DBRule);
        try {
            tableSe2DBHandler.appendSE2DWTask(pTableSe2DBRule.getParallelSize());
        } catch (Exception ex) {
            logger.warn("init mongodb worker for table " + pTableSe2DBRule.getTableName() + " unsuccessfully for " + ex.getMessage(), ex);
            tableSe2DBHandler = null;
        }
        try {
            tableSe2DBHandler.registerMsgHandler();
        } catch (Exception ex) {
            logger.warn("init metaq consumer for " + pTableSe2DBRule.getMqName() + " unsuccessfully for " + ex.getMessage(), ex);
            tableSe2DBHandler = null;
        }
        return tableSe2DBHandler;
    }

    private void appendSE2DWTask(int pTaskNum) throws Exception {
        for (int i = 0; i < pTaskNum; i++) {
            SE2DWWorker se2DBWorker = null;
            try {
                se2DBWorker = SE2DWWorkerFactory.getSE2DBWroker(this, i);
                se2DBWorkerSet.add(se2DBWorker);
                threadPool.execute(se2DBWorker);
            } catch (Exception ex) {
                logger.error("execute serializing to db " + tableSe2DBRule.getDbType() + " is failed for " + ex.getMessage(), ex);
                throw ex;
            }
        }
        logger.info(tableSe2DBRule.getTableName() + "," + tableSe2DBRule.getConnStr() + "," + tableSe2DBRule.getUserName() + "," + tableSe2DBRule.getPassword());
    }

    private void stopSE2DBTask() {
        for (SE2DWWorker se2DBWorker : se2DBWorkerSet) {
            se2DBWorker.shutdown();
        }
    }

    public void incLoadVolume(int pLoadVolume) {
        loadVolume.addAndGet(pLoadVolume);
    }

    public long getLoadVolume() {
        return loadVolume.longValue();
    }

    public static void startupConsumer() throws Exception {
        Iterator consumerItor = consumerSet.keySet().iterator();
        while (consumerItor.hasNext()) {
            String key = (String) consumerItor.next();
            REReceiveSession consumer = consumerSet.get(key);
            consumer.start();
            logger.info("consumer " + key + " has started successfully");
        }
    }

    public static void shutdownConsumer() throws REConnectionException {
        Iterator consumerItor = consumerSet.keySet().iterator();
        while (consumerItor.hasNext()) {
            String key = (String) consumerItor.next();
            REReceiveSession consumer = consumerSet.get(key);
            consumer.shutdown();
            logger.info("consumer " + key + " has stopped successfully");
        }
    }

    public void stop() {
        logger.info("stop tableSe2" + tableSe2DBRule.getDbType().toUpperCase() + "DWHandler for table " + tableSe2DBRule.getTableName() + " using mq " + tableSe2DBRule.getMqName() + "...");
        int dataCacheSize = -1;
        while (true) {
            dataCacheSize = dataCache.size();
            if (dataCacheSize < 1) {
                logger.info("all data of " + tableSe2DBRule.getTableName() + " has been cleared");
                break;
            } else {
                logger.info("waiting  data of " + tableSe2DBRule.getTableName() + " to be cleared " + dataCacheSize);
                try {
                    Thread.sleep(3000);
                } catch (Exception ex) {
                }
            }
        }
        logger.info("stop se2" + tableSe2DBRule.getDbType() + "dw worker for table " + tableSe2DBRule.getTableName() + " ...");
        stopSE2DBTask();
        threadPool.shutdown();
        try {
            threadPool.awaitTermination(10, TimeUnit.SECONDS);
        } catch (Exception ex) {
            logger.warn("stop thread pool unsuccessfully for " + ex.getMessage(), ex);
        }
        synchronized (consumerLock) {
            REReceiveSession consumer = consumerSet.get(tableSe2DBRule.getsGroup());
            tableSe2DWHandlerSet.remove(tableSe2DBRule.getsGroup());
        }
        logger.info("stop tableSe2" + tableSe2DBRule.getDbType().toUpperCase() + "DWHandler for table " + tableSe2DBRule.getTableName() + " successfully");
    }

    private void registerMsgHandler() throws Exception {

        synchronized (consumerLock) {
            REReceiveSession consumer = consumerSet.get(tableSe2DBRule.getsGroup());
            if (consumer == null) {
                REConnection conn = new REConnection(RuntimeEnv.getParamAsString(RuntimeEnv.NAME_SERVER_LIST));
                REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(this.getMQName());
                logger.info("rocketmqpull from " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM));
                if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("last")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
                } else if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("first")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
                } else if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("time")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
                    String ftime = RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM_Time);
                    try {
                        long ft = Integer.valueOf(ftime);
                        logger.info("pull data from mq and time is " + timeMillisToHumanString3(ft));
                        builder.setConsumeTimestamp(ft);
                    } catch (Exception E) {
                        long ft = System.currentTimeMillis() - 1800000;
                        builder.setConsumeTimestamp(ft);
                        logger.info("not get time from conf_rocketMQPullFromTime and is " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM_Time) + "and pull data from last 30 mints ago and time is " + timeMillisToHumanString3(ft));
                    }

                } else {
                    logger.warn("rocketmqpull from   last unknow is " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM));
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
                }
                builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
                    @Override
                    public boolean handle(REMessage message) {
//                        TableSe2DWHandler tableSe2DWHandler = tableSe2DWHandlerSet.get(tableSe2DBRule.getTableName());
                        TableSe2DWHandler tableSe2DWHandler = tableSe2DWHandlerSet.get(tableSe2DBRule.getsGroup());
                        tableSe2DWHandler.disposeMessages(message);
                        logger.info("pushdata " + tableSe2DBRule.getTableName() + " mq is  " + tableSe2DBRule.getMqName());
                        return true;
                    }
                });
                builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
                    @Override
                    public boolean handle(byte[] message) {
                        logger.error("formatter fuilure ! size = " + message.length);
                        return true;
                    }
                });

                builder.setGroupName("iie_di_" + RuntimeEnv.getParam(RuntimeEnv.REGION) + "_" + tableSe2DBRule.getsGroup());
                consumer = (REReceiveSession) builder.build();
                logger.info("init consumer for sgroup " + tableSe2DBRule.getsGroup() + " successfully");
                consumerSet.put(tableSe2DBRule.getsGroup(), consumer);
            }
            tableSe2DWHandlerSet.put(tableSe2DBRule.getsGroup(), this);
        }
        logger.info("register handler of mq " + this.getTableSe2DBRule().getMqName() + " for " + this.getTableSe2DBRule().getTableName() + " successfully");
    }

    public void disposeMessages(REMessage msg) {
        try {
            logger.info(Thread.currentThread().getId() + " receives message " + msg.data.size() + " at " + System.nanoTime() + " docschma " + msg.docSchemaName + " from " + tableSe2DBRule.getMqName());
            try {
                String tabname = tableSe2DBRule.getTableName();
                if (ROCKETMQ_DOC_DESC.get(tabname) == null) {
                    HashMap<String, REFieldType> newtable = new HashMap<String, REFieldType>();
                    logger.info("rule name is " + tableSe2DBRule.getRuleName());
                    for (int i = 0; i < msg.colNames.length; i++) {
                        if (RuntimeEnv.TABLE_RC_COLUMNS.get(tableSe2DBRule.getRuleName()).get(msg.colNames[i]) == null) {
                            logger.debug("rocketmqColumns compare  configColumns  and cfc is null  " + msg.colNames[i]);
                        }
                        newtable.put(msg.colNames[i], msg.types[i]);//mq中的字段。和字段类型
                    }
                    for (Map.Entry<String, String> entry : RuntimeEnv.TABLE_RC_COLUMNS.get(tableSe2DBRule.getRuleName()).entrySet()) {
                        String rname = entry.getKey();
                        if (newtable.get(rname) == null) {
                            logger.debug("configColumns compare rocketmqColumns and roc is null " + rname);
                        }
                    }
                    logger.info("docSchemaName " + msg.docSchemaName + " desc HashMap " + newtable);
                    ROCKETMQ_DOC_DESC.put(tabname, newtable);
                }
                if (msg.data.size() < 1) {
                    logger.warn("doc set is empty");
                    return;
                }
            } catch (Exception ex) {
                logger.error("parse docs unsuccessfully for " + ex.getMessage(), ex);
                return;
            } finally {

            }
            List<HashMap> recordSet = new ArrayList<HashMap>();

            int colLen = msg.colNames.length;
            for (Object[] doc : msg.data) {
                HashMap<String, Object> hm = new HashMap<String, Object>();
                for (int i = 0; i < colLen; ++i) {
                    if (doc[i] == null) {
                        hm.put(msg.colNames[i], null);
                        continue;
                    }
                    switch (msg.types[i]) {
                        case Strings:
                            hm.put(msg.colNames[i], doc[i]);
                            break;
                        case String:
                        default:
                            hm.put(msg.colNames[i], doc[i]);
                    }
                }
                recordSet.add(hm);
            }
            int bufferTime = 0;
            while (true) {
                logger.debug("dataCache size:" + dataCache.size());
                if (!dataCache.offer(recordSet)) {
                    if(bufferTime++ > 20){
                        logger.info("put to " + tableSe2DBRule.getTableName() + " buffer fail,will retry ...");
                        bufferTime = 0;
                    }
                    try {
                        Thread.sleep(200);
                    } catch (Exception ex) {
                    }
                } else {
                    logger.info("analysisSize " + recordSet.size() + "receMsgSize " + msg.data.size());
                    break;
                }
            }
        } catch (Exception ex) {
            logger.fatal("unknown exception " + ex.getMessage(), ex);
        }
    }

    public List getData() {
        try {
            return dataCache.poll();
        } catch (Exception ex) {
            return null;
        }
    }

    public Executor getExecutor() {
        return null;
    }

    public TableSe2DBRule getTableSe2DBRule() {
        return tableSe2DBRule;
    }

    public String getMQName() {
        return tableSe2DBRule.getMqName();
    }

    public String getTableName() {
        return tableSe2DBRule.getTableName();
    }

    public int getBatchSize() {
        return tableSe2DBRule.getBatchSize();
    }

    public String getRuleName() {
        return tableSe2DBRule.getRuleName();
    }

    public static String timeMillisToHumanString3(final long t) {
        Calendar cal = Calendar.getInstance();
        cal.setTimeInMillis(t);
        return String.format("%04d%02d%02d%02d%02d%02d",//
                cal.get(Calendar.YEAR),//
                cal.get(Calendar.MONTH) + 1,//
                cal.get(Calendar.DAY_OF_MONTH),//
                cal.get(Calendar.HOUR_OF_DAY),//
                cal.get(Calendar.MINUTE),//
                cal.get(Calendar.SECOND));
    }
}
